package com.wulaobo.state;

import com.wulaobo.constant.Constants;
import com.wulaobo.functions.WaterSensorMapFunction;
import com.wulaobo.pojo.WaterSensor;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;

/**
 * @Author wuwenbo
 * @Date 2024/7/10 11:11
 * @ClassName: KeyedListStateDemo
 * @Description: 案例:针对每种传感器输出最高的3个水位值
 */
public class KeyedListStateDemo {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        SingleOutputStreamOperator<WaterSensor> sensorDS = env
                .socketTextStream(Constants.IP, 7777)
                .map(new WaterSensorMapFunction());

        sensorDS.keyBy(WaterSensor::getId).process(new KeyedProcessFunction<String, WaterSensor, String>() {

            // TODO 1.定义状态
            ListState<Integer> listState;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);

                // TODO 2.在open方法中，初始化状态
                // 状态描述器两个参数：第一个参数，起个名字，不重复；第二个参数，存储的类型
                listState = getRuntimeContext().getListState(new ListStateDescriptor<Integer>("listState", Types.INT));
            }

            @Override
            public void processElement(WaterSensor value, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                List<Integer> list = new ArrayList<>();
                Iterable<Integer> iterable = listState.get();

                for (Integer vc:iterable) {
                    list.add(vc);
                }
                list.add(value.getVc());
                list.sort(new Comparator<Integer>() {
                    @Override
                    public int compare(Integer o1, Integer o2) {
                        return o2 - o1;
                    }
                });

                int size = list.size();
                int length = size > 3 ? 3 : size;
                List<Integer> subList = list.subList(0, length);
                out.collect("key:" + value.getId() + "subList：" + subList);
                listState.update(subList);
            }

        }).print();


        env.execute();

    }

}
